﻿using System;
using System.Collections.Concurrent;
using System.IO;
using System.Linq;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Peer
{
    public class PeerStream : IDisposable
    {
        protected readonly CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();

        protected Socket Socket { get; set; }
        public Guid Id { get; private set; }
        protected virtual bool CanSend { get { return Socket != null && Socket.Connected; } }


        private readonly ConcurrentDictionary<Guid, NotifyLock<PeerResponse>> notifyHandler = new ConcurrentDictionary<Guid, NotifyLock<PeerResponse>>();

        private readonly ConcurrentDictionary<Guid, MemoryStream> concurrentDictionary = new ConcurrentDictionary<Guid, MemoryStream>();
        /// <summary>
        /// 转换器
        /// </summary>
        internal readonly BaseConverter converter;

        public event OnReceiveEvent OnReceive;

        public event EventHandler<Exception> OnError;

        public event EventHandler<PeerStream> OnDisconnect;


        /// <summary>
        /// 头长度 guid 16 ,packageType 1, (bodyLength + isend)  4
        /// </summary>
        private const int HEAD_LENGTH = 16 + 1 + 4;

        /// <summary>
        /// 单包最大长度 1 M
        /// </summary>
        private const int MAX_BODY_LENGTH = 1 << 20;

        private const int MAX_LENGTH = MAX_BODY_LENGTH + HEAD_LENGTH;

        internal PeerStream(Guid id, Socket socket, BaseConverter converter)
        {
            this.Id = id;
            this.Socket = socket;
            this.converter = converter;
        }

        internal void ReceiveAsync()
        {
            while (Socket.Connected && !cancellationTokenSource.IsCancellationRequested)
            {
                Receive();
            }
        }

        private void Receive()
        {
            byte[] header = new byte[HEAD_LENGTH];
            Socket.ReceiveTimeout = 0;
            int headerLength = Socket.Receive(header);
            if (headerLength == 0)
            {
                throw new SocketException((int)SocketError.ConnectionAborted);
            }
            if (headerLength != header.Length)
            {
                // 非头, 丢了
                return;
            }
            Guid id = new Guid(header.AsSpan().Slice(0, 16));
            var packageType = (PackageType)header[16];
            int bodyLength = BitConverter.ToInt32(header, 17);
            bool isEnd = (bodyLength & 1) == 1;
            bodyLength >>= 1;
            switch (packageType)
            {
                case PackageType.Void:
                    ReadVoid(id);
                    break;
                case PackageType.Response:
                    ReadResponse(id, null, packageType);
                    break;
                default:
                    ReadDataBefore(id, packageType, isEnd, bodyLength);
                    break;
            }
        }

        private void ReadDataBefore(Guid guid, PackageType packageType, bool isEnd, int bodyLength)
        {
            if (bodyLength == 0)
            {
                Start(guid, packageType, new MemoryStream());
                return;
            }
            var bytes = new ArraySegment<byte>(new byte[bodyLength]);
            int length = Socket.Receive(bytes);
            if (length != bodyLength)
            {
                CallError(new IndexOutOfRangeException("body length is error"));
                return;
            }
            var memoryStream = concurrentDictionary.GetOrAdd(guid, guid => new MemoryStream());
            // 尝试创建的时候直接开启，没找到合适的方法
            memoryStream.Write(bytes);
            if (isEnd)
            {
                concurrentDictionary.TryRemove(guid, out _);
                memoryStream.Position = 0;
                Start(guid, packageType, memoryStream);
            }
        }

        #region Read


        protected virtual void ReadVoid(Guid guid)
        {

        }

        private void ReadError(Guid guid, Stream stream)
        {
            using (stream)
            {
                OnError?.Invoke(this, (Exception)converter.DeserializeObject(stream));
            }
        }


        private void ReadResponse(Guid guid, Stream stream, PackageType packageType)
        {
            if (notifyHandler.TryGetValue(guid, out var notifyLock))
            {
                notifyLock.Notify(new PeerResponse(converter, guid, packageType, stream));
            }
        }

        private void ReadData(Guid guid, Stream stream, bool withResponse)
        {
            using (stream)
            {
                var request = new PeerRequest(converter, guid, stream);
                PeerResponse? response;
                try
                {
                    response = OnReceive?.Invoke(this, request);
                }
                catch (Exception e)
                {
                    if (withResponse)
                    {
                        // 发送错误
                        Send(request.CreateResponse(e));
                    }
                    return;
                }
                if (withResponse)
                {
                    // 发送响应
                    Send(response ?? request.CreateResponse());
                }
            }
        }

        #endregion

        #region Send
        public Task SendAsync(Stream stream)
        {
            return SendData(Guid.NewGuid(), stream, PackageType.Data);
        }

        public Task<PeerResponse> SendWithResponseAsync(Stream stream)
        {
            return Task.Factory.StartNew(() =>
            {
                var guid = Guid.NewGuid();
                try
                {
                    NotifyLock<PeerResponse> notifyLock = new NotifyLock<PeerResponse>();
                    notifyHandler.TryAdd(guid, notifyLock);
                    SendData(guid, stream, PackageType.DataWithResponse).GetAwaiter().GetResult();
                    if (!notifyLock.Wait(60000))
                        throw new TimeoutException("Sss SendWithResponseAsync Timeout");
                    return notifyLock.Value;
                }
                finally
                {
                    notifyHandler.TryRemove(guid, out _);
                }
            }, TaskCreationOptions.PreferFairness);
        }

        internal void Send(PeerResponse response)
        {
            switch (response.PackageType)
            {
                case PackageType.Response:
                    SendVoid(response.Id, true);
                    break;
                case PackageType.ResponseError:
                case PackageType.ResponseData:
                    SendData(response.Id, response.Data, response.PackageType).GetAwaiter().GetResult();
                    break;
                default:
                    break;
            }
        }

        protected void SendVoid(Guid guid, bool isRespone)
        {
            try
            {
                var guidBytes = guid.ToByteArray();
                byte[] header = new byte[HEAD_LENGTH];
                Array.Copy(guidBytes, header, 16);
                header[16] = (byte)(isRespone ? PackageType.Response : PackageType.Void);
                Socket.Send(header);
            }
            catch (Exception e)
            {
                CallError(e);
            }
        }


        private async Task SendData(Guid guid, Stream stream, PackageType package)
        {
            try
            {
                bool isEnd = false;
                while (!isEnd)
                {
                    byte[] bytes = new byte[MAX_LENGTH];
                    int bodyLength = await stream.ReadAsync(bytes, HEAD_LENGTH, MAX_BODY_LENGTH);
                    // 实际长度小于，最大长度
                    isEnd = bodyLength < MAX_BODY_LENGTH;
                    var header = BuilderHeader(guid, package, bodyLength, isEnd);
                    Array.Copy(header, 0, bytes, 0, header.Length);
                    Socket.Send(bytes, HEAD_LENGTH + bodyLength, SocketFlags.None);
                }
            }
            catch (Exception e)
            {
                CallError(e);
            }
            finally
            {
                stream.Dispose();
            }
        }
        #endregion


        private byte[] BuilderHeader(Guid guid, PackageType packageType, int bodyLength, bool isEnd = false)
        {
            var guidBytes = guid.ToByteArray();
            byte[] header = new byte[HEAD_LENGTH];
            Array.Copy(guidBytes, header, 16);
            header[16] = (byte)packageType;
            bodyLength <<= 1;
            if (isEnd) bodyLength |= 1; // 结尾标识
            byte[] lengthBytes = BitConverter.GetBytes(bodyLength);
            Array.Copy(lengthBytes, 0, header, 17, lengthBytes.Length);
            return header;
        }

        private void Start(Guid guid, PackageType packageType, Stream stream)
        {
            Task.Factory.StartNew(() =>
            {
                try
                {
                    switch (packageType)
                    {
                        case PackageType.Error:
                            ReadError(guid, stream);
                            break;
                        case PackageType.Data:
                        case PackageType.DataWithResponse:
                            ReadData(guid, stream, packageType == PackageType.DataWithResponse);
                            break;
                        case PackageType.ResponseData:
                        case PackageType.ResponseError:
                            ReadResponse(guid, stream, packageType);
                            break;
                        default:
                            stream.Dispose();
                            break;
                    }
                }
                catch (Exception e)
                {
                    CallError(e);
                }
                finally
                {
                    concurrentDictionary.TryRemove(guid, out _);
                }
            }, TaskCreationOptions.PreferFairness);
        }

        ~PeerStream()
        {
            Dispose();
        }

        internal void CallError(Exception exception)
        {
            OnError?.Invoke(this, exception);
        }

        public void Close(bool trigger = true)
        {
            if (Socket.Connected)
            {
                Socket.Shutdown(SocketShutdown.Both);
                Socket.Close();
            }
            if (trigger)
                OnDisconnect?.Invoke(this, this);
        }

        public virtual void Dispose()
        {
            cancellationTokenSource.Cancel();
            Close(true);
            GC.SuppressFinalize(this);
        }
    }
}
